package pers.zyl.amqp;

import com.rabbitmq.client.*;

import java.io.*;
import java.nio.charset.StandardCharsets;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.logging.SimpleFormatter;
import java.util.regex.Pattern;

public class Rabbitmq {
    private static final String QUEUE_NAME = "queue_demo";
    private static final String IP_ADDRESS = "192.168.31.4";
    private static final int PORT = 5672;
    private static final String EXCHANGE_NAME = "exchange_demo";
    private static final String ROUTING_KEY = "routingkey_demo";
    private static final String USERNAME = "root";
    private static final String PASSWORD = "root";


    /**
     * 延迟队列
     */
    private static final String exchangeDlx = "exchange.dlx";
    private static final String exchangeNormal = "exchange.normal";
    private static final String queueDlx = "queue.dlx";
    private static final String queueNormal = "queue.normal";
    private static final String bindNormalRoutingKey = "normal";

    public static Connection getConnection() throws IOException, TimeoutException {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Rabbitmq.IP_ADDRESS);
        connectionFactory.setPassword(PASSWORD);
        connectionFactory.setUsername(USERNAME);
        connectionFactory.setPort(PORT);
        return connectionFactory.newConnection();
    }

    public static Channel getChannel(Connection connection) throws IOException {
        return connection.createChannel();
    }


    public static void initDelayQueue(Channel channel, SortedSet<Long> sortedSet) throws IOException {
        channel.confirmSelect();
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("Nack, SeqNo: " + deliveryTag + ", multi ple: " + multiple);
                long toElement = deliveryTag + 1;
                if (multiple) {
                    sortedSet.headSet(toElement + 1).clear();
                } else {
                    sortedSet.remove(toElement);
                }
            }

            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                long toElement = deliveryTag + 1;
                if (multiple) {
                    sortedSet.headSet(toElement + 1).clear();
                } else {
                    sortedSet.remove(toElement);
                }
                //消息重发
            }
        });
        //创建一个死信交换机
        channel.exchangeDeclare(exchangeDlx, BuiltinExchangeType.FANOUT.getType(), true, false, null);
        //创建一个普通交换机
        HashMap<String, Object> arguments = new HashMap<>();
        channel.exchangeDeclare(exchangeNormal, BuiltinExchangeType.DIRECT.getType(), true, false, arguments);
        //创建一个死信队列
        channel.queueDeclare(queueDlx, true, false, false, null);
        channel.queueBind(queueDlx, exchangeDlx, bindNormalRoutingKey);
        //创建一个普通队列
        HashMap<String, Object> arguments1 = new HashMap<>();
        arguments1.put("x-message-ttl", 30 * 1000);
        arguments1.put("x-dead-letter-exchange", exchangeDlx);
//        arguments1.put("x-dead-ler", bindNormalRoutingKey);
        channel.queueDeclare(queueNormal, true, false, false, arguments1);
        channel.queueBind(queueNormal, exchangeNormal, bindNormalRoutingKey);
    }

    public static void sendMessage(Channel channel, String exchangeName,String routingKey, String message, SortedSet<Long> sortedSet) throws IOException {
        if (message == null) {
            DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            message = dateFormat.format(new Date());
        }
        channel.basicPublish(exchangeName, routingKey,
                MessageProperties.MINIMAL_PERSISTENT_BASIC,
                message.getBytes(StandardCharsets.UTF_8));
        long nextPublishSeqNo = channel.getNextPublishSeqNo();
        System.out.println(nextPublishSeqNo + "已发送");
        sortedSet.add(nextPublishSeqNo);
    }

    public static void consumeMessage(Channel channel, String queueName) throws IOException {
        channel.basicQos(64);
        channel.basicConsume(queueName, false, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println(Thread.currentThread().getName());
                System.out.println(new String(body, StandardCharsets.UTF_8));
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        });
    }

    private static class RabbitProducer {
        private static final String AlternateExchangeName = "AlternateExchange";
        private static final String AlternateQueue = "AlternateQueue";

        public static void createAlternateExchange(Channel channel) throws IOException {
            final HashMap<String, Object> arguments = new HashMap<>();
            arguments.put("alternate-exchange", "myAe");
            //创建备份交换机
            channel.exchangeDeclare(AlternateExchangeName, BuiltinExchangeType.FANOUT, true, false, false, null);
            //创建交换机(持久化,不自动删除)
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT.getType(), true, false, arguments);
            //创建备份队列
//            channel.queueDeclare(AlternateQueue, )
        }

        public static Channel createMessageToQueue(String message, SortedSet<Long> sortedSet) throws IOException, TimeoutException {
            ConnectionFactory factory = new ConnectionFactory();

            factory.setHost(IP_ADDRESS);
            factory.setPort(PORT);
            factory.setUsername(USERNAME);
            factory.setPassword(PASSWORD);
            final Connection connection = factory.newConnection();
            final Channel channel = connection.createChannel();
            //创建一个type="direct"、持久化的、非自动删除的交换器
            channel.exchangeDeclare(EXCHANGE_NAME, "direct", true, false, null);
            //创建一个持久化、非排他的、非自动删除的队列
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            //绑定队列
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);

            if (message == null) {
                message = "hello the word";
            }
            channel.confirmSelect();
            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    System.out.println("Nack, SeqNo: " + deliveryTag + ", multi ple: " + multiple);
                    long toElement = deliveryTag + 1;
                    if (multiple) {
                        sortedSet.headSet(toElement + 1).clear();
                    } else {
                        sortedSet.remove(toElement);
                    }
                }

                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    long toElement = deliveryTag + 1;
                    if (multiple) {
                        sortedSet.headSet(toElement + 1).clear();
                    } else {
                        sortedSet.remove(toElement);
                    }
                    //消息重发
                }
            });

            return channel;
        }
        /**
         * 发送数据
         * @param channel
         * @param message
         * @param sortedSet
         * @throws IOException
         */
        public static void sendMessage(Channel channel, String message, SortedSet<Long> sortedSet) throws IOException {
            DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            message = dateFormat.format(new Date());
            channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY,
                    MessageProperties.MINIMAL_PERSISTENT_BASIC,
                    message.getBytes(StandardCharsets.UTF_8));
            long nextPublishSeqNo = channel.getNextPublishSeqNo();
            System.out.println(nextPublishSeqNo + "已发送");
            sortedSet.add(nextPublishSeqNo);
        }
    }



    private static class RabbitmqConsumer {
        public static Channel Consumed() throws IOException, TimeoutException {
            final boolean[] isConsumed = {false};

            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost(IP_ADDRESS);
            factory.setPort(PORT);
            factory.setUsername(USERNAME);
            factory.setPassword(PASSWORD);
            final Connection connection = factory.newConnection();
            final Channel channel = connection.createChannel();
            channel.basicQos(64);
            final Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    final String s = new String(body, StandardCharsets.UTF_8);
                    System.out.println(s);
                    try {
                        TimeUnit.SECONDS.sleep(2);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    channel.basicAck(envelope.getDeliveryTag(), false);
                    isConsumed[0] = true;
                }
            };
            channel.basicConsume(QUEUE_NAME, consumer);
            return channel;
        }
    }

    public static void testDelayQueue() throws IOException, TimeoutException {

        Connection connection = Rabbitmq.getConnection();
        Channel productChannel = Rabbitmq.getChannel(connection);
        Channel consumerChannel = Rabbitmq.getChannel(connection);

        TreeSet<Long> sortedSet = new TreeSet<>();
        Rabbitmq.initDelayQueue(productChannel, sortedSet);
        BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(System.in));

//        Rabbitmq.consumeMessage(consumerChannel, Rabbitmq.queueDlx);
        final String end = "end";
        final String send = "send";
        final String consumed = "consumed";
        while (true) {

            String s = bufferedReader.readLine();

            if (end.equals(s)) {
                break;
            } else if (send.equals(s)) {
                Rabbitmq.sendMessage(productChannel, Rabbitmq.exchangeNormal, Rabbitmq.bindNormalRoutingKey, null, sortedSet);
            }
        }

    }

    public static void main(String[] args) throws InterruptedException, IOException, TimeoutException {
        Rabbitmq.testDelayQueue();
    }
}
